package com.fwmagic.spark.core.cases.groupwithtopn

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
  * 分组TopN:统计每门学科中访问次数排在前3名的老师
  * 1、对学科和老师组合进行分组统计数量
  * 2、取出前3个
  *
  * 方式一：聚合后对Key分组，获取所有的Vlaue(迭代器)，toList后排序取TopN
  */
object FavoriteTeacher1 {
    def main(args: Array[String]): Unit = {
        val isLocal = args(0)

        val conf = new SparkConf().setAppName(this.getClass.getSimpleName)

        if (isLocal.toBoolean) {
            conf.setMaster("local[*]")
        }

        val sc = new SparkContext(conf)

        //读取数据
        val lines: RDD[String] = sc.textFile(args(1))

        //切分数据
        /* http://bigdata.fwmagic.com/huangzhong */
        val subjectTeacherAndOne: RDD[((String, String), Int)] = lines.map(line => {
            val words: Array[String] = line.split("/")
            val subject = words(2).split("\\.")(0)
            val teacher = words(3)
            ((subject, teacher), 1)
        })

        //聚合计算
        val reduced: RDD[((String, String), Int)] = subjectTeacherAndOne.reduceByKey(_ + _)

        //全局排序，不符合要求，想要分组排序
//        val sorted: RDD[((String, String), Int)] = reduced.sortBy(_._2,false)

        //
        /**
          * 1、按照学科进行分组，得到每个迭代器中的学科是一样的
          *
          * (javaee,List(((javaee,machao),5), ((javaee,sunquan),5), ((javaee,zhangfei),3)))
          * (python,List(((python,lvbu),4), ((python,zhugeliang),3), ((python,sunshangxiang),2)))
          * (bigdata,List(((bigdata,liubei),6), ((bigdata,guanyu),4), ((bigdata,zhaoyun),2)))
          *
          * 缺点：如果一个组中的数据量过多，可能会造成Driver端内存溢出，不适用于数据量大的场景！
          */
        val grouped: RDD[(String, Iterable[((String, String), Int)])] = reduced.groupBy(tp => tp._1._1)

        val topN = args(2).toInt

        //2、再对迭代器中数据进行排序取TopN
        val result: RDD[(String, List[((String, String), Int)])] = grouped.mapValues(_.toList.sortBy(-_._2).take(topN))

        //输出
        result.foreach(println)

        sc.stop()
    }
}

/*
预期计算结果：
(javaee,machao,5)
(javaee,sunquan,5)
(javaee,zhangfei,3)
(javaee,sunshangxiang,2)
-- (javaee,caocao,1)

(bigdata,liubei,6)
(bigdata,guanyu,4)
(bigdata,zhaoyun,2)
-- (bigdata,huangzhong,1)

(python,lvbu,4)
(python,zhugeliang,3)
(python,liushan,2)
-- (python,diaochan,1)

准备数据：
http://bigdata.fwmagic.com/huangzhong
http://bigdata.fwmagic.com/zhaoyun
http://bigdata.fwmagic.com/zhaoyun
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/liubei
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://bigdata.fwmagic.com/guanyu
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/zhangfei
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/machao
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/sunquan
http://javaee.fwmagic.com/caocao
http://javaee.fwmagic.com/sunshangxiang
http://javaee.fwmagic.com/sunshangxiang
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/lvbu
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/zhugeliang
http://python.fwmagic.com/liushan
http://python.fwmagic.com/liushan
http://python.fwmagic.com/diaochan
*/